/*
 * Copyright (C) 2014 Square, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package okio;

import static okio.Util.checkOffsetAndCount;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.concurrent.TimeUnit;

import javax.annotation.Nullable;

/**
 * This timeout uses a background thread to take action exactly when the timeout
 * occurs. Use this to implement timeouts where they aren't supported natively,
 * such as to sockets that are blocked on writing.
 *
 * <p>
 * Subclasses should override {@link #timedOut} to take action when a timeout
 * occurs. This method will be invoked by the shared watchdog thread so it
 * should not do any long-running operations. Otherwise we risk starving other
 * timeouts from being triggered.
 *
 * <p>
 * Use {@link #sink} and {@link #source} to apply this timeout to a stream. The
 * returned value will apply the timeout to each operation on the wrapped
 * stream.
 *
 * <p>
 * Callers should call {@link #enter} before doing work that is subject to
 * timeouts, and {@link #exit} afterwards. The return value of {@link #exit}
 * indicates whether a timeout was triggered. Note that the call to
 * {@link #timedOut} is asynchronous, and may be called after {@link #exit}.
 */
public class AsyncTimeout extends Timeout {
	/**
	 * Don't write more than 64 KiB of data at a time, give or take a segment.
	 * Otherwise slow connections may suffer timeouts even when they're making
	 * (slow) progress. Without this, writing a single 1 MiB buffer may never
	 * succeed on a sufficiently slow connection.
	 */
	// 一次不要写超过64 KiB的数据
	private static final int TIMEOUT_WRITE_SIZE = 64 * 1024;

	/** Duration for the watchdog thread to be idle before it shuts itself down. */
	private static final long IDLE_TIMEOUT_MILLIS = TimeUnit.SECONDS.toMillis(60);
	private static final long IDLE_TIMEOUT_NANOS = TimeUnit.MILLISECONDS.toNanos(IDLE_TIMEOUT_MILLIS);

	/**
	 * The watchdog thread processes a linked list of pending timeouts, sorted in
	 * the order to be triggered. This class synchronizes on AsyncTimeout.class.
	 * This lock guards the queue.
	 *
	 * <p>
	 * Head's 'next' points to the first element of the linked list. The first
	 * element is the next node to time out, or null if the queue is empty. The head
	 * is null until the watchdog thread is started and also after being idle for
	 * {@link #IDLE_TIMEOUT_MILLIS}.
	 */
	// Head的“next”指向链表的第一个元素。第一个元素是超时的下一个节点，如果队列为空，则为null。
	static @Nullable AsyncTimeout head;

	/** True if this node is currently in the queue. */
	private boolean inQueue;

	/** The next node in the linked list. */
	private @Nullable AsyncTimeout next;

	/** If scheduled, this is the time that the watchdog should time this out. */
	// 超时时间
	private long timeoutAt;

	// 使用者在操作之前首先要调用enter方法，这样相当于注册了这个超时监听，然后配对的实现exit方法，这个exit有一个返回值会表明超时是否触发，请注意这个timeout是异步的，可能会在exit后才调用
	public final void enter() {
		// 这里只是做了判断以及设置inQueue的状态，真正的是调用 scheduleTimeout 方法来加入到链表中
		if (inQueue)
			throw new IllegalStateException("Unbalanced enter/exit");
		long timeoutNanos = timeoutNanos();
		boolean hasDeadline = hasDeadline();
		if (timeoutNanos == 0 && !hasDeadline) {
			return; // No timeout and no deadline? Don't bother with the queue. 没有超时的设置
		}
		inQueue = true;
		scheduleTimeout(this, timeoutNanos, hasDeadline);
	}

	private static synchronized void scheduleTimeout(AsyncTimeout node, long timeoutNanos, boolean hasDeadline) {
		// 启动看门狗线程，并在触发第一次超时时创建head节点。
		if (head == null) {
			head = new AsyncTimeout();
			new Watchdog().start();
		}

		long now = System.nanoTime();
		if (timeoutNanos != 0 && hasDeadline) {
			// Compute the earliest event; either timeout or deadline. Because nanoTime can
			// wrap around,
			// Math.min() is undefined for absolute values, but meaningful for relative
			// ones.
			// 计算最早的事件;要么超时，要么截止。因为nanoTime可以封装，所以Math.min()对于绝对值没有定义，但是对于相对值有意义。
			node.timeoutAt = now + Math.min(timeoutNanos, node.deadlineNanoTime() - now);
		} else if (timeoutNanos != 0) {
			// 超时时间是当前时间+timeoutNanos
			node.timeoutAt = now + timeoutNanos;
		} else if (hasDeadline) {
			// 超时时间就是deadlineNanoTime
			node.timeoutAt = node.deadlineNanoTime();
		} else {
			throw new AssertionError();
		}

		// Insert the node in sorted order.
		long remainingNanos = node.remainingNanos(now);
		for (AsyncTimeout prev = head; true; prev = prev.next) {
			// 如果下一个为null或者剩余时间比下一个短 就插入node(当前对象)
			// 可以看出这个链表实际上是按照剩余的超时时间来进行排序的，快到超时的节点排在表头，依次往后递增
			if (prev.next == null || remainingNanos < prev.next.remainingNanos(now)) {
				node.next = prev.next;
				prev.next = node;
				if (prev == head) {
					AsyncTimeout.class.notify(); // Wake up the watchdog when inserting at the front.
				}
				break;
			}
		}
	}

	// 如果超时返回true
	public final boolean exit() {
		if (!inQueue)
			return false;
		inQueue = false;
		return cancelScheduledTimeout(this);
	}

	/** Returns true if the timeout occurred. */
	private static synchronized boolean cancelScheduledTimeout(AsyncTimeout node) {
		// Remove the node from the linked list.
		for (AsyncTimeout prev = head; prev != null; prev = prev.next) {
			// 找到了就没有超时，同时node出队
			if (prev.next == node) {
				prev.next = node.next;
				node.next = null;
				return false;
			}
		}

		// The node wasn't found in the linked list: it must have timed out!
		return true;
	}

	/**
	 * Returns the amount of time left until the time out. This will be negative if
	 * the timeout has elapsed and the timeout should occur immediately.
	 */
	// 返回在超时之前剩余的时间量。如果超时已经过去，并且应该立即发生超时，则该值为负。
	private long remainingNanos(long now) {
		return timeoutAt - now;
	}

	/**
	 * Invoked by the watchdog thread when the time between calls to
	 * {@link #enter()} and {@link #exit()} has exceeded the timeout.
	 */
	protected void timedOut() {
	}

	/**
	 * Returns a new sink that delegates to {@code sink}, using this to implement
	 * timeouts. This works best if {@link #timedOut} is overridden to interrupt
	 * {@code sink}'s current operation.
	 */
	public final Sink sink(final Sink sink) {
		return new Sink() {
			@Override
			public void write(Buffer source, long byteCount) throws IOException {
				checkOffsetAndCount(source.size, 0, byteCount);

				while (byteCount > 0L) {
					// Count how many bytes to write. This loop guarantees we split on a segment
					// boundary.
					long toWrite = 0L;
					for (Segment s = source.head; toWrite < TIMEOUT_WRITE_SIZE; s = s.next) {
						int segmentSize = s.limit - s.pos;
						toWrite += segmentSize;
						if (toWrite >= byteCount) {
							toWrite = byteCount;
							break;
						}
					}

					// Emit one write. Only this section is subject to the timeout.
					boolean throwOnTimeout = false;
					enter();
					try {
						sink.write(source, toWrite);
						byteCount -= toWrite;
						throwOnTimeout = true;
					} catch (IOException e) {
						throw exit(e);
					} finally {
						exit(throwOnTimeout);
					}
				}
			}

			@Override
			public void flush() throws IOException {
				boolean throwOnTimeout = false;
				enter();
				try {
					sink.flush();
					throwOnTimeout = true;
				} catch (IOException e) {
					throw exit(e);
				} finally {
					exit(throwOnTimeout);
				}
			}

			@Override
			public void close() throws IOException {
				boolean throwOnTimeout = false;
				enter();
				try {
					sink.close();
					throwOnTimeout = true;
				} catch (IOException e) {
					throw exit(e);
				} finally {
					exit(throwOnTimeout);
				}
			}

			@Override
			public Timeout timeout() {
				return AsyncTimeout.this;
			}

			@Override
			public String toString() {
				return "AsyncTimeout.sink(" + sink + ")";
			}
		};
	}

	/**
	 * Returns a new source that delegates to {@code source}, using this to
	 * implement timeouts. This works best if {@link #timedOut} is overridden to
	 * interrupt {@code sink}'s current operation.
	 */
	public final Source source(final Source source) {
		return new Source() {
			@Override
			public long read(Buffer sink, long byteCount) throws IOException {
				boolean throwOnTimeout = false;
				enter();
				try {
					long result = source.read(sink, byteCount);
					throwOnTimeout = true;
					return result;
				} catch (IOException e) {
					throw exit(e);
				} finally {
					exit(throwOnTimeout);
				}
			}

			@Override
			public void close() throws IOException {
				boolean throwOnTimeout = false;
				try {
					source.close();
					throwOnTimeout = true;
				} catch (IOException e) {
					throw exit(e);
				} finally {
					exit(throwOnTimeout);
				}
			}

			@Override
			public Timeout timeout() {
				return AsyncTimeout.this;
			}

			@Override
			public String toString() {
				return "AsyncTimeout.source(" + source + ")";
			}
		};
	}

	/**
	 * Throws an IOException if {@code throwOnTimeout} is {@code true} and a timeout
	 * occurred. See {@link #newTimeoutException(java.io.IOException)} for the type
	 * of exception thrown.
	 */
	final void exit(boolean throwOnTimeout) throws IOException {
		boolean timedOut = exit();
		if (timedOut && throwOnTimeout)
			throw newTimeoutException(null);
	}

	/**
	 * Returns either {@code cause} or an IOException that's caused by {@code cause}
	 * if a timeout occurred. See {@link #newTimeoutException(java.io.IOException)}
	 * for the type of exception returned.
	 */
	final IOException exit(IOException cause) throws IOException {
		if (!exit())
			return cause;
		return newTimeoutException(cause);
	}

	/**
	 * Returns an {@link IOException} to represent a timeout. By default this method
	 * returns {@link java.io.InterruptedIOException}. If {@code cause} is non-null
	 * it is set as the cause of the returned exception.
	 */
	protected IOException newTimeoutException(@Nullable IOException cause) {
		InterruptedIOException e = new InterruptedIOException("timeout");
		if (cause != null) {
			e.initCause(cause);
		}
		return e;
	}

	private static final class Watchdog extends Thread {
		Watchdog() {
			super("Okio Watchdog");
			setDaemon(true);
		}

		@Override
		public void run() {
			while (true) {
				try {
					AsyncTimeout timedOut;
					synchronized (AsyncTimeout.class) {
						timedOut = awaitTimeout();

						// 没有找到一个node来interrupt 继续
						if (timedOut == null)
							continue;

						// 队列完全是空的。让这个线程退出，并在下一次调用scheduleTimeout()时创建另一个看门狗线程。
						if (timedOut == head) {
							head = null;
							return;
						}
					}

					timedOut.timedOut();
				} catch (InterruptedException ignored) {
				}
			}
		}
	}

	/**
	 * Removes and returns the node at the head of the list, waiting for it to time
	 * out if necessary. This returns {@link #head} if there was no node at the head
	 * of the list when starting, and there continues to be no node after waiting
	 * {@code IDLE_TIMEOUT_NANOS}. It returns null if a new node was inserted while
	 * waiting. Otherwise this returns the node being waited on that has been
	 * removed.
	 */
	// 删除并返回列表顶部的节点，如有必要，等待它超时。如果在开始时列表的头部没有节点，并且在等待{@code
	// IDLE_TIMEOUT_NANOS}之后仍然没有节点，则返回{@link #head}。
	// 如果在等待时插入了新节点，则返回null。否则，它将返回被等待的已被删除的节点。
	static @Nullable AsyncTimeout awaitTimeout() throws InterruptedException {
		// 拿到下一个节点
		AsyncTimeout node = head.next;

		// 如果queue为空，等待直到有node进队，或者触发IDLE_TIMEOUT_MILLS
		if (node == null) {
			long startNanos = System.nanoTime();
			AsyncTimeout.class.wait(IDLE_TIMEOUT_MILLIS);
			return head.next == null && (System.nanoTime() - startNanos) >= IDLE_TIMEOUT_NANOS ? head // The idle
																										// timeout
																										// elapsed.
																										// 空闲超时过期
					: null; // The situation has changed. 情况已经改变了
		}

		long waitNanos = node.remainingNanos(System.nanoTime());

		// 这个node依然还没有超时，继续等待
		if (waitNanos > 0) {
			// Waiting is made complicated by the fact that we work in nanoseconds,
			// but the API wants (millis, nanos) in two arguments.
			long waitMillis = waitNanos / 1000000L;
			waitNanos -= (waitMillis * 1000000L);
			AsyncTimeout.class.wait(waitMillis, (int) waitNanos);
			return null;
		}

		// 这个node已经超时，就删除
		head.next = node.next;
		node.next = null;
		return node;
	}
}
